package com.ajun.listener;

import com.ajun.common.constant.RedisConstants;
import com.ajun.common.rabbit.QueueConstants;
import com.ajun.common.util.RedisUtil;
import com.ajun.entity.dto.OrderDto;
import com.ajun.service.DispatchService;
import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.concurrent.TimeUnit;

/**
 * 消费订单 MQ
 *
 * @author LiJun
 * @date 2021-07-15
 */
@Slf4j
@Service
public class DispatchListener {

    @Resource
    private DispatchService dispatchService;

    @Resource
    private RedisTemplate<Object, Object> redisTemplate;

    /**
     * 消息消费过程中如果出现异常, RabbitMQ会一直进行消息重试
     * 解决方案:
     * 1.控制重试次数
     * 2.try-catch + 手动 ack
     * 3.try-catch + 手动 ack + 死信队列(防止消息丢失)
     */
    @RabbitListener(queues = {QueueConstants.ORDER_DISPATCH_QUEUE})
    public void messageConsumer(@Payload String message,
                                Channel channel,
                                CorrelationData correlationData,
                                @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws IOException {
        try {
            // 获取消息
            log.info("收到MQ消息是: {}", message);
            // 反序列化 json
            OrderDto orderDto = JSON.parseObject(message, OrderDto.class);

            // 业务幂等处理
            String key = RedisConstants.ORDER_IDM + orderDto.getOrderId();
            if (RedisUtil.instance().getCacheObject(key) == null) {
                // 派单处理(需要幂等处理，防止重复消费)
                dispatchService.dispatch(orderDto);
                RedisUtil.instance().setCacheObject(key, orderDto, 10L, TimeUnit.MINUTES);
            }

            // 模拟异常
            System.out.println(1 / 0);

            // 手动 ack 告诉 MQ 消息已经正常消费
            channel.basicAck(deliveryTag, false);
        } catch (Exception ex) {
            /*
                如果出现异常,要根据实际情况去进行重发,
                重发一次后,丢失还是存库根据业务场景决定
                param1: 消息的 tag; param2:多条处理; param3: requeue(重发)
                false不会重发,但会把消息打入死信队列
                true 会死循环进行重发,建议如果 true,不要加 try/catch 否则会造成死循环
             */
            // 手动 nack,出现异常,消息不重发,会将消息转移到绑定的死信队列中
            log.error("出现异常...消息转移到死信队列...");
            channel.basicNack(deliveryTag, false, false);
        }
    }
}
